package com.lzwk.app.ods.wkForm;

import com.alibaba.fastjson.JSONObject;
import com.lzwk.app.ods.adb.AdbCustomerDeserialization;
import com.lzwk.beans.ocpx.WkFormAccountInfo;
import com.lzwk.common.test.TestConstant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: WkCDCTest
 * @Author Yeaser
 * @Date: 2022/4/7 16:14
 * @Version 1.0
 * Description: TODO
 */
public class WkCDCTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        MySqlSource<String> mySqlSourceHistory = MySqlSource.<String>builder()
                .hostname(TestConstant.TESTHOST)
                .port(Integer.parseInt(TestConstant.TESTPORT))
                .username(TestConstant.TESTUSERNAME)
                .password(TestConstant.TESTPASSWORD)
                .databaseList(TestConstant.TESTWKFORMDB)
                .tableList(TestConstant.TESTWKFORMDB + "." + TestConstant.TESTWKFORMCDCACCOUNTINFO)
                .deserializer(new AdbCustomerDeserialization("common"))
                .startupOptions(StartupOptions.initial())
                .includeSchemaChanges(true)
                .splitSize(80000)
                .build();

        SingleOutputStreamOperator<WkFormAccountInfo> source1 = env
                .fromSource(mySqlSourceHistory, WatermarkStrategy
                                .noWatermarks(),
                        "MySQL WK Source")
                .process(new ProcessFunction<String, WkFormAccountInfo>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<WkFormAccountInfo> out) throws Exception {
                        WkFormAccountInfo wkFormAccountInfo = null;
                        JSONObject data = JSONObject.parseObject(value);
                        if (data.getJSONObject("data").toJSONString() != null) {
                            JSONObject dataDetail = data.getJSONObject("data");
                            wkFormAccountInfo = JSONObject.parseObject(dataDetail.toJSONString(), WkFormAccountInfo.class);
                        }
                        out.collect(wkFormAccountInfo);
                    }
                });


        Table wkAccountInfoTable = tenv.fromDataStream(source1,
                $("id"), $("gender"), $("telephone"), $("career"), $("age"), $("wechat_nickname"), $("create_time"),
                $("create_time").rowtime().as("watermark_time"), $("pt").proctime());

        tenv.createTemporaryView("wk_form_account_info", wkAccountInfoTable);


        tenv.executeSql(" CREATE TABLE adb_class_student ( " +
                "  `id` INT, " +
                "  `division_activity_id` INT, " +
                "  `activity_term_id` INT, " +
                "  `term_class_id` INT, " +
                "  `account_id` INT, " +
                "  `state` STRING, " +
                "  `refund_status` INT, " +
                "  `remark` STRING, " +
                "  `create_time` TIMESTAMP(0), " +
                "  `update_time` TIMESTAMP(0), " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url' = 'jdbc:mysql://" + TestConstant.TESTHOST + ":" + TestConstant.TESTPORT + "/" + TestConstant.TESTADBDB + "?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true&tinyInt1isBit=false&transformedBitIsBoolean=false', " +
                "  'table-name' = " + "'" + TestConstant.TESTADBJDBCDIVISIONCLASSSTUDENT + "'" + ", " +
                "  'username'=" + "'" + TestConstant.TESTUSERNAME + "'" + ", " +
                "  'password'=" + "'" + TestConstant.TESTPASSWORD + "'" + ", " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        tenv.executeSql(                " select `account_id`, " +
                "        `class_id`, " +
                "        `parent_activity_id`, " +
                "        `activity_id`, " +
                "        `parent_term_id`, " +
                "        `term_id`, " +
                "        `student_state`, " +
                "        `refund_status`, " +
                "        student_create_time, " +
                "        `liveroom_id`, " +
                "        `target_roi`, " +
                "        `activity_name`, " +
                "        `activity_state`, " +
                "        `term_rank`, " +
                "        `term_start_time`, " +
                "        `term_end_time`, " +
                "        `term_student_cnt`, " +
                "        `term_state`, " +
                "        `term_promote_start_time`, " +
                "        `class_channel_category`, " +
                "        `class_name`, " +
                "        `class_rank`, " +
                "        `class_object_id`, " +
                "        `class_object_type`, " +
                "        `class_student_number`, " +
                "        `class_can_redirect`, " +
                "        `class_state`, " +
                "        `class_monitor_id`, " +
                "        `class_monitor_name`, " +
                "        `age`, " +
                "        `career`, " +
                "        `city`, " +
                "        `headimgurl`, " +
                "        `city_level`, " +
                "        `acc_sex`, " +
                "        `province`, " +
                "        `create_time`, " +
                "        `source_type`, " +
                "        `acc_nickname`, " +
                "        `wk_acc_nickname`, " +
                "        `input_cost`, " +
                "        `acc_phone_num`, " +
                "        `wk_acc_phone_num`, " +
                "        `wk_acc_sex`, " +
                "        `op` " +
                "        from( " +
                " SELECT ifnull(student.`account_id`, -1) AS `account_id`, " +
                "       ifnull(student.`term_class_id`,-1)                                                      AS class_id, " +
                "       cast(null as int)                                                                       AS `parent_activity_id`, " +
                "       cast(null as int)                                                                       AS activity_id, " +
                "       cast(null as int)                                                                       AS `parent_term_id`, " +
                "       cast(null as int)                                                                       AS term_id, " +
                "       cast(null as string)                                                                    AS student_state, " +
                "       cast(null as int)                                                                       AS refund_status, " +
                "       cast(null as string)                                                                    AS student_create_time, " +
                "       cast(null as int)                                                                       AS `liveroom_id`, " +
                "       cast(null as string)                                                                    AS `activity_name`, " +
                "       cast(null as string)                                                                    AS activity_state, " +
                "       cast(null as int)                                                                       AS term_rank, " +
                "       cast(null as string)                                                                    AS term_start_time, " +
                "       cast(null as string)                                                                    AS `term_end_time`, " +
                "       cast(null as int)                                                                       AS term_student_cnt, " +
                "       cast(null as string)                                                                    AS term_state, " +
                "       cast(null as string)                                                                    AS `term_promote_start_time`, " +
                "       cast(null as string)                                                                    AS class_channel_category, " +
                "       cast(null as string)                                                                    AS class_name, " +
                "       cast(null as int)                                                                       AS class_rank, " +
                "       cast(null as int)                                                                       AS class_object_id, " +
                "       cast(null as string)                                                                    AS class_object_type, " +
                "       cast(null as int)                                                                       AS class_student_number, " +
                "       cast(null as int)                                                                       AS class_can_redirect, " +
                "       cast(null as string)                                                                    AS class_state, " +
                "       cast(null as int)                                                                       AS `class_monitor_id`, " +
                "       cast(null as string)                                                                    AS `class_monitor_name`, " +
                "       ifnull(acc_info.age,'')                                                                 AS age, " +
                "       ifnull(regexp_replace(acc_info.career,'\\n|\\t',''),'')                                 AS career, " +
                "       cast(null as string)                                                                    AS city, " +
                "       cast(null as string)                                                                    AS headimgurl, " +
                "       cast(null as string)                                                                    AS city_level, " +
                "       cast(null as string)                                                                    AS acc_sex,  " + //"性别:未知0 男1 女2"
                "       cast(null as string)                                                                    AS province, " +
                "       cast(null as string)                                                                    AS create_time, " +
                "       cast(null as string)                                                                    AS source_type, " +
                "       cast(null as string)                                                                    AS acc_nickname, " +
                "       ifnull(regexp_replace(acc_info.wechat_nickname,'\\n|\\t',''),'')                        AS wk_acc_nickname, " +
                "       cast(null as string)                                                                    AS acc_phone_num, " +
                "       ifnull(regexp_replace(acc_info.telephone,'\\n|\\t',''),'')                              AS wk_acc_phone_num, " +
                "       ifnull(cast(acc_info.gender as INT),0)                                                  AS wk_acc_sex, " +
                "       cast(null as int)                                                                       AS op, " +
                "       cast(null as double)                                                                    AS input_cost, " +
                "       cast(null as double)                                                                    AS target_roi " +
                "       FROM wk_form_account_info acc_info " +
                "       JOIN adb_class_student  FOR SYSTEM_TIME AS OF `acc_info`.`pt` AS student " +
                "              ON acc_info.id = student.account_id " +
                ")").print();

    }
}
